package com.haruhi.event;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.net.JarURLConnection;
import java.net.URL;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;

/**
 * 默认服务端和默认客户端来的消息都会收集到队列 这两可以自定义 参考 com.haruhi.example.MyClientEndpoint 和 com.haruhi.example.MyServerEndpoint
 * 统一消费（消息=发送给所有handler类 广播模式 每条消息都会发给所有handler）
 * 每个handler每次执行handle独有一个线程
 */
public class MessageConsumer {

    private static final AtomicBoolean started = new AtomicBoolean(false);
    public static boolean started() {
        return started.get();
    }
    private static ExecutorService executor;

    private static final List<MessageHandler> handlers = new ArrayList<>();

    public synchronized static void start(String... packageName) throws IOException, ClassNotFoundException {
        if(packageName == null || packageName.length == 0){
            return;
        }
        if(!started.get()){
            scanAndRegister(packageName);
            initPool(handlers.size() > 0 ? handlers.size() : 1);
            new Thread(()->{
                while (true){
                    try {
                        Message message = MessageQueue.take();
                        for (MessageHandler handler : handlers) {
                            executor.execute(()-> {
                                try {
                                    handler.handle(message);
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            });
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }).start();
            started.set(true);
        }
    }

//    private static void scanAndRegister(String pack){
//        Reflections reflections = new Reflections(pack);
//        Set<Class<? extends MessageHandler>> subTypesOf = reflections.getSubTypesOf(MessageHandler.class);
//        for (Class<? extends MessageHandler> aClass : subTypesOf) {
//            try {
//                MessageHandler messageHandler = aClass.newInstance();
//                handlers.add(messageHandler);
//            }catch (Exception e){
//                e.printStackTrace();
//            }
//        }
//    }
    private static void scanAndRegister(String... packs) throws IOException, ClassNotFoundException {
        Set<String> pkgSet = new HashSet<>(Arrays.asList(packs));
        for (String pack : pkgSet) {
            List<Class<?>> allClassByInterface = getAllClassByInterface(pack, MessageHandler.class);
            for (Class<?> aClass : allClassByInterface) {
                try {

                    MessageHandler messageHandler =(MessageHandler) aClass.newInstance();
                    handlers.add(messageHandler);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }

    }

    public static List<Class<?>> getAllClassByInterface(String packageName,Class<?> c) throws IOException, ClassNotFoundException {
        List<Class<?>> returnClassList = null;
        if (c.isInterface()) {
            // 获取当前的包名
            // 获取当前包下以及子包下所以的类
            List<Class<?>> allClass = getClasses(packageName);
            if (allClass != null) {
                returnClassList = new ArrayList<Class<?>>();
                for (Class<?> cls : allClass) {
                    // 判断是否是同一个接口
                    if (c.isAssignableFrom(cls)) {
                        // 本身不加入进去
                        if (!c.equals(cls)) {
                            returnClassList.add(cls);
                        }
                    }
                }
            }
        }

        return returnClassList;
    }
    private synchronized static void initPool(int poolSize){
        if(executor == null){
            executor = new EnhancedThreadPoolExecutor(poolSize, poolSize * 2 + 1, 60,
                    TimeUnit.SECONDS, false, new ThreadFactory() {

                private final AtomicInteger threadCount = new AtomicInteger();
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setDaemon(false);
                    thread.setName("pool-consumer-"+threadCount.incrementAndGet());
                    return thread;
                }
            });
        }
    }


    /**
     * 常规线程池执行任务规则： corePoolSize -> queueSize -> maximumPoolSize -> RejectedExecutionHandler
     *      corePoolSize未满，且有空闲线程，也会创建新线程
     * 该线程池执行任务规则： corePoolSize -> maximumPoolSize -> queueSize -> RejectedExecutionHandler
     *      corePoolSize未满，且有空闲线程，不会创建新线程，而是使用空闲线程去执行任务
     */
    static class EnhancedThreadPoolExecutor implements ExecutorService {

        private final ThreadPoolExecutor delegate;

        EnhancedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, boolean allowCoreThreadTimeOut, ThreadFactory threadFactory){
            delegate = new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,new Queue(this),threadFactory);
            delegate.allowCoreThreadTimeOut(allowCoreThreadTimeOut);
        }

        public int getActiveCount(){
            return delegate.getActiveCount();
        }
        public int getPoolSize(){
            return delegate.getPoolSize();
        }
        public BlockingQueue<Runnable> getQueue(){
            return delegate.getQueue();
        }
        public int getCorePoolSize(){
            return delegate.getCorePoolSize();
        }
        public int getMaximumPoolSize(){
            return delegate.getMaximumPoolSize();
        }


        @Override
        public void execute(Runnable runnable){
            if(getActiveCount() < getPoolSize()){
                if(getQueue().offer(runnable)){
                    return;
                }
            }
            delegate.execute(runnable);
        }

        @Override
        public void shutdown() {
            delegate.shutdown();
        }

        @Override
        public List<Runnable> shutdownNow() {
            return delegate.shutdownNow();
        }

        @Override
        public boolean isShutdown() {
            return delegate.isShutdown();
        }

        @Override
        public boolean isTerminated() {
            return delegate.isTerminated();
        }

        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
            return delegate.awaitTermination(timeout,unit);
        }

        @Override
        public <T> Future<T> submit(Callable<T> task) {
            return delegate.submit(task);
        }

        @Override
        public <T> Future<T> submit(Runnable task, T result) {
            return delegate.submit(task,result);
        }

        @Override
        public Future<?> submit(Runnable task) {
            return delegate.submit(task);
        }

        @Override
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
            return delegate.invokeAll(tasks);
        }

        @Override
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
            return delegate.invokeAll(tasks,timeout,unit);
        }

        @Override
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
            return delegate.invokeAny(tasks);
        }

        @Override
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            return delegate.invokeAny(tasks,timeout,unit);
        }
    }

    static class Queue extends LinkedBlockingQueue<Runnable> {
        private final EnhancedThreadPoolExecutor parent;
        Queue(EnhancedThreadPoolExecutor executor){
            this.parent = executor;
        }
        @Override
        public boolean offer(Runnable runnable) {
            if (parent.getPoolSize() < parent.getMaximumPoolSize()) {
                return false;
            }
            return super.offer(runnable);
        }
    }

    public static List<Class<?>> getClasses(String packageName) throws IOException, ClassNotFoundException {
        // 第一个class类的集合
        List<Class<?>> classes = new ArrayList<Class<?>>();
        // 是否循环迭代
        boolean recursive = true;
        // 获取包的名字 并进行替换
        String packageDirName = packageName.replace('.', '/');
        // 定义一个枚举的集合 并进行循环来处理这个目录下的things
        Enumeration<URL> dirs = Thread.currentThread().getContextClassLoader().getResources(packageDirName);
        // 循环迭代下去
        while (dirs.hasMoreElements()) {
            // 获取下一个元素
            URL url = dirs.nextElement();
            // 得到协议的名称
            String protocol = url.getProtocol();
            // 如果是以文件的形式保存在服务器上
            if ("file".equals(protocol)) {
                // 获取包的物理路径
                String filePath = URLDecoder.decode(url.getFile(), "UTF-8");
                // 以文件的方式扫描整个包下的文件 并添加到集合中
                findAndAddClassesInPackageByFile(packageName, filePath, recursive, classes);
            } else if ("jar".equals(protocol)) {
                // 如果是jar包文件
                // 定义一个JarFile
                JarFile jar = ((JarURLConnection) url.openConnection()).getJarFile();
                // 从此jar包 得到一个枚举类
                Enumeration<JarEntry> entries = jar.entries();
                // 同样的进行循环迭代
                while (entries.hasMoreElements()) {
                    // 获取jar里的一个实体 可以是目录 和一些jar包里的其他文件 如META-INF等文件
                    JarEntry entry = entries.nextElement();
                    String name = entry.getName();
                    // 如果是以/开头的
                    if (name.charAt(0) == '/') {
                        // 获取后面的字符串
                        name = name.substring(1);
                    }
                    // 如果前半部分和定义的包名相同
                    if (name.startsWith(packageDirName)) {
                        int idx = name.lastIndexOf('/');
                        // 如果以"/"结尾 是一个包
                        if (idx != -1) {
                            // 获取包名 把"/"替换成"."
                            packageName = name.substring(0, idx).replace('/', '.');
                        }
                        // 如果可以迭代下去 并且是一个包
                        if ((idx != -1) || recursive) {
                            // 如果是一个.class文件 而且不是目录
                            if (name.endsWith(".class") && !entry.isDirectory()) {
                                // 去掉后面的".class" 获取真正的类名
                                String className = name.substring(packageName.length() + 1, name.length() - 6);
                                try {
                                    // 添加到classes
                                    classes.add(Class.forName(packageName + '.' + className));
                                } catch (ClassNotFoundException e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    }
                }
            }
        }
        return classes;
    }

    public static void findAndAddClassesInPackageByFile(String packageName, String packagePath, final boolean recursive, List<Class<?>> classes) throws ClassNotFoundException {
        // 获取此包的目录 建立一个File
        File dir = new File(packagePath);
        // 如果不存在或者 也不是目录就直接返回
        if (!dir.exists() || !dir.isDirectory()) {
            return;
        }
        // 如果存在 就获取包下的所有文件 包括目录
        File[] dirfiles = dir.listFiles(new FileFilter() {
            // 自定义过滤规则 如果可以循环(包含子目录) 或则是以.class结尾的文件(编译好的java类文件)
            @Override
            public boolean accept(File file) {
                return (recursive && file.isDirectory()) || (file.getName().endsWith(".class"));
            }
        });
        // 循环所有文件
        for (File file : dirfiles) {
            // 如果是目录 则继续扫描
            if (file.isDirectory()) {
                findAndAddClassesInPackageByFile(packageName + "." + file.getName(), file.getAbsolutePath(), recursive, classes);
            } else {
                // 如果是java类文件 去掉后面的.class 只留下类名
                String className = file.getName().substring(0, file.getName().length() - 6);
                classes.add(Class.forName(packageName + '.' + className));
            }
        }
    }
}
